fix(socket-mode): only release connect_operation_lock when this task acquired it - #1926
fix(socket-mode): only release connect_operation_lock when this task acquired it#1926ckarnell wants to merge 2 commits into
Conversation
AsyncBaseSocketModeClient.connect_to_new_endpoint() releases the lock whenever connect_operation_lock.locked() is true, rather than when this coroutine actually acquired it. asyncio.Lock has no notion of ownership, so release() from a task that never held the lock succeeds and frees it for everyone. That is reachable on an ordinary reconnect. connect() cancels message_receiver and current_session_monitor on every successful reconnection, and both of those tasks call connect_to_new_endpoint(). So one of them can be cancelled while suspended inside acquire(), and its finally block then releases the lock belonging to the reconnect still in progress, dropping mutual exclusion around it. Track acquisition in a local instead, which is what the synchronous BaseSocketModeClient already does with its acquired flag.
|
Thanks for the contribution! Before we can merge this, we need @ckarnell to sign the Salesforce Inc. Contributor License Agreement. |
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #1926 +/- ##
==========================================
- Coverage 84.17% 84.15% -0.03%
==========================================
Files 118 118
Lines 13425 13427 +2
==========================================
- Hits 11301 11299 -2
- Misses 2124 2128 +4 ☔ View full report in Codecov by Harness. |
|
Hi I've signed the CLA, can this please get reviewed? |
WilliamBergamin
left a comment
There was a problem hiding this comment.
Hello 👋 thanks for your thoughtful contribution 🙏
I left a few comments mainly around aligning the logic with the sync implementation
| await self.connect_operation_lock.acquire() | ||
| acquired = True |
There was a problem hiding this comment.
Could we mirror the same pattern found ion the sync implementation
| await self.connect_operation_lock.acquire() | |
| acquired = True | |
| acquired = await self.connect_operation_lock.acquire(blocking=True, timeout=5) |
| acquired = True | ||
| if self.trace_enabled: | ||
| self.logger.debug(f"For reconnection, the connect_operation_lock was acquired (session: {session_id})") | ||
| if force or not await self.is_connected(): |
There was a problem hiding this comment.
Similar here
| if force or not await self.is_connected(): | |
| if force or (acquired and not await self.is_connected()): |
| await self.connect_operation_lock.acquire() | ||
| acquired = True | ||
| if self.trace_enabled: | ||
| self.logger.debug(f"For reconnection, the connect_operation_lock was acquired (session: {session_id})") |
There was a problem hiding this comment.
| self.logger.debug(f"For reconnection, the connect_operation_lock was acquired (session: {session_id})") | |
| self.logger.debug(f"For reconnection, the connect_operation_lock was acquired: {acquired} (session: {session_id})") |
| @@ -0,0 +1,70 @@ | |||
| import asyncio | |||
There was a problem hiding this comment.
Could we mirror the same testing pattern found in the sync implementation 🙏
Review feedback on slackapi#1926: mirror connect_to_new_endpoint's sync pattern, so the acquire is bounded and the reconnect is gated on having got the lock. asyncio.Lock.acquire() takes no arguments, unlike threading.Lock, so the sync client's acquire(blocking=True, timeout=5) is spelled with asyncio.wait_for here. The suggested acquire(blocking=True, timeout=5) raises TypeError on an asyncio lock. Adds a test for the branch this introduces: when the lock cannot be acquired, the endpoint is not rotated, connect() is not called, and no lock is released. The existing test passed either way, so the new behaviour was uncovered.
|
Thanks for the review. The There was no coverage for the bounded acquire, so I added a test. When the lock cannot be acquired, the endpoint stays un-rotated and On mirroring the sync test pattern: I could not find a sync lock test to mirror, and the sync socket-mode tests are integration-style against a mock server. Did you have a specific test in mind, or is the async unit test above sufficient here? |
Summary
AsyncBaseSocketModeClient.connect_to_new_endpoint()releasesconnect_operation_lockwhenever the lock is merelylocked(), rather than when this coroutine actually acquired it:asyncio.Lockhas no notion of ownership, unlikethreading.Lock. Arelease()call from a task that never held the lock succeeds and frees it for everyone.Why this is reachable on an ordinary reconnect, not just at shutdown
connect()cancels both background tasks on every successful reconnection:and both of those tasks call
connect_to_new_endpoint()(aiohttp/__init__.py, inmonitor_current_session()andreceive_messages()). So:monitor_current_sessioncallsconnect_to_new_endpoint(), acquires the lock, and is insideawait self.connect().receive_messagesreaches its own reconnect path and suspends insideconnect_operation_lock.acquire().connect(),self.message_receiver.cancel()cancels the second task while it is waiting on the lock.CancelledErrorpropagates out ofacquire(), so that task never acquired anything, butlocked()isTruebecause the first task still holds it.finallyblock releases the first task's lock, mid-reconnect.Mutual exclusion around the reconnect is gone, so another reconnect can start while one is still in flight.
Reproduction
tests/slack_sdk_async/socket_mode/test_async_client_lock.pydrives the realAsyncBaseSocketModeClientwith aconnect()that blocks until released, cancels a second caller while it waits on the lock, and asserts the lock is still held.Without the change:
With the change it passes.
The fix
Track acquisition in a local, which is exactly what the synchronous
BaseSocketModeClientalready does:So this makes the async client consistent with its sync counterpart. Three lines, no behaviour change on the happy path: when the coroutine does acquire the lock,
acquiredisTrueand it releases exactly as before.What I verified
tests/slack_sdk_async/socket_mode/passes: 14 passed.black --line-length 125(the setting inpyproject.toml) leaves both files unchanged.main, not only in the released package.What I did not check
Whether this is the cause of any specific reported disconnection. The area has prior history (#1110, #1112) whose symptoms are consistent with losing mutual exclusion around reconnects, but I have not tied this to a particular report, and I have not tried to reproduce it against live Slack infrastructure.
A note on severity
Not a security issue, and it needs a cancellation to land in a narrow window, so it is unlikely to be a frequent cause of trouble. The consequence when it does happen is two concurrent reconnects rather than data loss. The sync client already gets this right, which is the main argument for the change.
Category
slack_sdk.socket_mode(Socket Mode client)Requirements
./scripts/run_validation.shafter making the changes. Not run in full. What I did run is listed under "What I verified" above: thetests/slack_sdk_async/socket_mode/suite (14 passed) andblack --line-length 125on both changed files. Happy to run the full script if you would like it before review.